package com.lgw.coupon.service.impl;

import cn.hutool.json.JSONUtil;
import com.lgw.commons.constant.KafkaConstant;
import com.lgw.commons.model.pojo.VoucherOrders;
import com.lgw.coupon.service.IKafkaService;
import com.lgw.coupon.service.IOrderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Slf4j
@Component
public class KafkaServiceImpl implements IKafkaService {
    @Autowired
    private IOrderService orderService;

    /**
     * 异步下单
     * kakfa自动调用消费消息
     * @param record
     */
    @Override
    @KafkaListener(topics = {KafkaConstant.TOPIC}, groupId = "coupon-1")
    public void consumeCouponKafkaMessage(ConsumerRecord<?, ?> record) {
        //当有消息从topic来的时候，kafka把消息反序列化为record，record会在这个方法里面被消费
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            //kafka里面的消息是String
            String message = (String)kafkaMessage.get();
            log.info("kafka客户端收到一个订单");
            //反序列化消息
            VoucherOrders voucherOrders = JSONUtil.toBean(message, VoucherOrders.class);
            try {
                orderService.generateOrder(voucherOrders);
                log.info("成功生成订单，userId={},goodsId={}",voucherOrders.getCustomerId(),voucherOrders.getVoucherId());
            } catch (Exception e) {
                log.error(e.getLocalizedMessage());
                log.info("生成订单时出现错误，userId={},goodsId={}",voucherOrders.getCustomerId(),voucherOrders.getVoucherId());
            }
        }else{
            log.warn("消息不存在");
        }
    }
}
